package producer;

import data.handle.PackMessage;
import data.store.DataWarehouse;
import data.store.LineItemSpace;
import data.store.PartSpace;
import local.LocalInstance;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.HashSet;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

public class BigFileReader {
    private int threadSize;
    private long fileLength1;
    private long fileLength2;
    private RandomAccessFile rAccessFile1;
    private RandomAccessFile rAccessFile2;
    private Set<StartEndPair> startEndPairs;
    private int partKeyIndex1;//part表零件号是第几个字段(part表中是0)
    private int partKeyIndex2;//lineitem表零件号是第几个字段(lineitem表中是1)
    private AtomicLong counter = new AtomicLong(0);

    private int instanceId;

    public BigFileReader(File file1, File file2, int threadSize, int partKeyIndex1, int partKeyIndex2, int instanceId) {
        this.fileLength1 = file1.length();
        this.fileLength2 = file2.length();
        this.threadSize = threadSize;
        this.partKeyIndex1 = partKeyIndex1;
        this.partKeyIndex2 = partKeyIndex2;
        try {
            this.rAccessFile1 = new RandomAccessFile(file1, "r");
            this.rAccessFile2 = new RandomAccessFile(file2, "r");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        startEndPairs = new HashSet<>();
        this.instanceId = instanceId;
    }

    public void start() {
        long everySize1 = this.fileLength1 / this.threadSize;
        long everySize2 = this.fileLength2 / this.threadSize;
        try {
            calculateStartEnd(0, 0, everySize1, everySize2);
        } catch (IOException e) {
            e.printStackTrace();
            return;
        }

        PartAndLineItemHandler.init(instanceId);//开始计时

        CountDownLatch downLatch = new CountDownLatch(threadSize);

        for (StartEndPair pair : startEndPairs) {
            System.out.println("分配分片：" + pair);
            new Thread(new SliceReaderTask(pair, new PartAndLineItemHandler(threadSize), downLatch)).start();
        }
    }

    private void calculateStartEnd(long start1, long start2, long size1, long size2) throws IOException {
        if (start1 > fileLength1 - 1) {
            return;
        }
        StartEndPair pair = new StartEndPair();
        pair.start1 = start1;
        pair.start2 = start2;
        long endPosition1 = start1 + size1 - 1;
        long endPosition2 = start2 + size2 - 1;

        if (endPosition1 >= fileLength1 - 1 || endPosition2 >= fileLength2 - 1) {
            pair.end1 = fileLength1 - 1;
            pair.end2 = fileLength2 - 1;
            startEndPairs.add(pair);
            return;
        }

        rAccessFile1.seek(endPosition1);
        rAccessFile2.seek(endPosition2);
        byte tmp1 = (byte) rAccessFile1.read();
        byte tmp2 = (byte) rAccessFile2.read();
        while (tmp1 != '\n' && tmp1 != '\r') {
            endPosition1++;
            if (endPosition1 >= fileLength1 - 1) {
                endPosition1 = fileLength1 - 1;
                break;
            }
            rAccessFile1.seek(endPosition1);
            tmp1 = (byte) rAccessFile1.read();
        }
        while (tmp2 != '\n' && tmp2 != '\r') {
            endPosition2++;
            if (endPosition2 >= fileLength2 - 1) {
                endPosition2 = fileLength2 - 1;
                break;
            }
            rAccessFile2.seek(endPosition2);
            tmp2 = (byte) rAccessFile2.read();
        }

        pair.end1 = tmp1 == '\r' ? endPosition1 + 1 : endPosition1;
        pair.end2 = tmp2 == '\r' ? endPosition2 + 1 : endPosition2;
        startEndPairs.add(pair);

        calculateStartEnd(pair.end1 + 1, pair.end2 + 1, size1, size2);//递归调用

    }

    private static class StartEndPair {
        public long start1;
        public long end1;
        public long start2;
        public long end2;

        @Override
        public String toString() {
            return "StartEndPair{" +
                    "start1=" + start1 +
                    ", end1=" + end1 +
                    ", start2=" + start2 +
                    ", end2=" + end2 +
                    '}';
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            StartEndPair that = (StartEndPair) o;
            return start1 == that.start1 && end1 == that.end1 && start2 == that.start2 && end2 == that.end2;
        }

        @Override
        public int hashCode() {
            return Objects.hash(start1, end1, start2, end2);
        }
    }

    private class SliceReaderTask implements Runnable {
        private PartSpace partSpace;
        private LineItemSpace lineItemSpace;//用于存储在本地处理的那部分数据

        private long start1;
        private long sliceSize1;

        private long start2;
        private long sliceSize2;

        private Handler handler;

        private CountDownLatch countDownLatch;

        public SliceReaderTask(StartEndPair pair, Handler handler, CountDownLatch countDownLatch) {
            partSpace = DataWarehouse.allocationPartSpace();
            lineItemSpace = DataWarehouse.allocationLineItemSpace();

            this.start1 = pair.start1;
            this.sliceSize1 = pair.end1 - pair.start1 + 1;

            this.start2 = pair.start2;
            this.sliceSize2 = pair.end2 - pair.start2 + 1;

            this.handler = handler;

            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                long start = start1;
                long sliceSize = sliceSize1;
                int partKeyIndex = partKeyIndex1;
                RandomAccessFile rAccessFile = rAccessFile1;

                //对两个分片文件进行读取
                for (int i = 0; i < 2; i++) {
                    MappedByteBuffer mapBuffer = rAccessFile.getChannel().map(FileChannel.MapMode.READ_ONLY, start, sliceSize);
//                    byte[] line = new byte[256];//记录一行的数据
                    ByteBuffer byteBuffer = ByteBuffer.allocateDirect(256);//记录一行的数据
                    byteBuffer.put((byte) 0);
                    byteBuffer.put((byte) 0);//每个数据包头有两个字节表示该包数据的长度

                    byte[] partKey = new byte[7];//记录零件号
                    //下面三个数组只有在本地处理的数据才用得到
                    byte[] brand = new byte[8];//记录零件品牌
                    byte[] quantity = new byte[5];//记录零件数量
                    byte[] price = new byte[10];//记录零件价格

                    int key = 0;//转换后的零件号

                    short lineLength = 0;//一行数据的字节数
                    int partKeyLength = 0;//记录零件号在文件中的字节数
                    //下面三个变量只有在本地处理的数据才用得到
                    int curBrandIndex = 0;//记录零件品牌字节当前位置
                    int quantityLength = 0;//记录零件数量在文件中的字节数
                    int priceLength = 0;//记录零件价格在文件中的字节数

                    int curIndex = 0;//当前在第几个字段

                    int index = 0;//记录当前零件号的这一行数据会被后面哪个本地线程或远程端口处理

                    boolean partKeyFlag = true;//后续是否需要对零件号进行处理

                    boolean isLocalHandle = false;//是否是本地处理的数据

//                    int y = 1;
//                    boolean yFlag = true;

                    //开始以字节为单位进行读取
                    for (int n = 0; n < sliceSize; n++) {
                        byte b = mapBuffer.get();

                        //如果对零件号的字段读取完毕了，需要解析零件号
                        if(partKeyFlag && curIndex == partKeyIndex + 1){
                            partKeyFlag = false;//保证该过程只处理一次
                            String partKeyString = new String(partKey, 0, partKeyLength);
                            key = Integer.parseInt(partKeyString);//得到零件号
                            index = PartAndLineItemHandler.getIndex(key);//计算当前零件号的这一行数据会被后面哪个本地线程或远程端口处理
//                            if(key == 1){
//                                System.out.println("PartKey = 1");
//                                System.out.println("index = " + index);
//                            }
                            isLocalHandle = PartAndLineItemHandler.isLocalHandle(index);//是否会本地处理
                            if (isLocalHandle){//如果本地处理会将保存在byteBuffer中的字节转移到分配给本地的数据缓存空间
                                byteBuffer.flip();
                                byteBuffer.getShort();//移动指针到数据体开始的位置
                                while(byteBuffer.hasRemaining()){
                                    if(i == 0)
                                        partSpace.add(byteBuffer.get());
                                    else
                                        lineItemSpace.add(byteBuffer.get());
                                }
                                byteBuffer.clear();
                            }
                        }

                        //一行读取结束，进行数据分配
                        if (b == '\n') {
                            //如果是在本地处理的数据，会将解析出来的关键数据分发给本地的数据处理线程进行处理
                            if (isLocalHandle) {
                                PackMessage packMessage = new PackMessage();
                                packMessage.setPartKey(key);
                                if(i == 0){//part表的处理
                                    partSpace.add((byte) '\r');
                                    partSpace.add((byte) '\n');//一行数据存储结束
                                    //转换后的零件品牌
                                    int brandIndex = Integer.parseInt(new String(brand, 6, 2));
                                    packMessage.setBrand(brandIndex);
                                    curBrandIndex = 0;
                                }else {//lineItem表的处理
                                    lineItemSpace.add((byte) '\r');
                                    lineItemSpace.add((byte) '\n');//一行数据存储结束
                                    //转换后的零件数量
                                    int l_quantity = Integer.parseInt(new String(quantity, 0, quantityLength));
                                    //转换后的零件价格
                                    double extendedPrice = Double.parseDouble(new String(price, 0, priceLength));
                                    packMessage.setQuantity(l_quantity);
                                    packMessage.setPrice(extendedPrice);
                                    quantityLength = 0;
                                    priceLength = 0;
                                }

                                handler.localHandle(packMessage, PartAndLineItemHandler.getLocalIndex(index), i);//交给本地线程处理数据

                            } else {//远程处理的数据
                                byteBuffer.putShort(0, lineLength);//将一行数据的字节数保存在数据包头位置

                                handler.remoteHandle(byteBuffer, index);//交给远程实例处理数据

                                //为下一行数据读取做准备
                                byteBuffer.clear();
                                byteBuffer.put((byte) 0);
                                byteBuffer.put((byte) 0);//每个数据包头有两个字节表示该包数据的长度
                            }
                            //为下一行数据读取做准备
                            curIndex = 0;
                            lineLength = 0;
                            partKeyLength = 0;
                            partKeyFlag = true;
                            isLocalHandle = false;
//                            y++;
//                            yFlag = true;
                            continue;
                        } else if (b == '\r') {
                            continue;
                        } else if (b == '|') {//当前字段的位置加1
                            curIndex++;
                        } else if (partKeyIndex == curIndex) {
                            partKey[partKeyLength++] = b;
                        } else if(isLocalHandle && i == 0 && curIndex == 3){
                            brand[curBrandIndex++] = b;
                        } else if(isLocalHandle && i == 1 && curIndex == 4){
                            quantity[quantityLength++] = b;
                        } else if(isLocalHandle && i == 1 && curIndex == 5){
                            price[priceLength++] = b;
                        }

                        if (isLocalHandle && i == 0) //本地处理的数据，part表
                            partSpace.add(b);
                        else
                            if (isLocalHandle && i == 1) //本地处理的数据，lineItem表
                                lineItemSpace.add(b);
                            else {
//                                if(yFlag && start == 0 && i == 1 && y<=25){
//                                    byteBuffer.put((byte) 0);
//                                    yFlag = false;
//                                }
                                byteBuffer.put(b);//记录远程处理的数据
                                lineLength++;
                            }
                    }

                    if(i == 0){//part表结束
                        handler.localFile1End();
//                        handler.remoteFile1End();
                        countDownLatch.countDown();
                    } else {//lineItem表结束
                        handler.localFile2End();
//                        handler.remoteFile2End();
                    }
                    //进行lineItem表的读取准备
                    start = start2;
                    sliceSize = sliceSize2;
                    partKeyIndex = partKeyIndex2;
                    rAccessFile = rAccessFile2;

                    countDownLatch.await();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
}
